package dk.almbrand.microservice.heartbeat;

import dk.almbrand.microservice.kafka.KafkaFacade;

import dk.almbrand.microservice.kafka.KafkaFacadeImpl;

import dk.almbrand.microservice.model.RequestEvent;

import dk.almbrand.microservice.model.ResponseEvent;

import dk.almbrand.microservice.util.GsonUtils;

import java.util.Date;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

public class HeartbeatImpl implements Heartbeat {

    public static final String PING = "ping";

    private Timer timer;
    private KafkaFacade facade;

    private boolean isAlive = false;

    public HeartbeatImpl(KafkaFacade facade) {
        this.facade = facade;
    }
    
    public Heartbeat build() {
        start();
        return this;
    }
    
    public void start() {
        timer = new Timer(); 
        timer.schedule(new HeartBeatTimer(), 60 * 1000); //delay in milliseconds        
    }
    
    public void stop() {
        if(timer != null) {
            timer.cancel();
        }
    }
    
    public void setIsAlive(boolean isAlive) {
        this.isAlive = isAlive;
    }

    public boolean isIsAlive() {
        return isAlive;
    }
    
    public static void main(String[] args) {
        //new HeartbeatImpl(new KafkaFacadeImpl("host", "Serializer")).start();
    }

    class HeartBeatTimer extends TimerTask {

        @Override
        public void run() {
            ResponseEvent event = new ResponseEvent();
            event.setObject(PING);
            facade.getProducer().publish("", event.getKey(), new GsonUtils<ResponseEvent>().toJSON(event));
        }
    }
}
